package com.bancomer.pig.functions;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.pig.LoadFunc;
import org.apache.pig.backend.hadoop.executionengine.mapReduceLayer.PigSplit;
import org.apache.pig.data.Tuple;
import org.apache.pig.data.TupleFactory;
import org.apache.hadoop.io.Text;

public class IndexOfLoader extends LoadFunc {
		
	
	@SuppressWarnings("rawtypes")
	private RecordReader reader;	
	private TupleFactory tupleFactory;	
	private List<IndexPojo> listIndexOf= new ArrayList<IndexPojo>();
	

	
	public IndexOfLoader(String ... indexLine){
		
		
		for(String colum:indexLine){
			String[] split=colum.split("-");
			IndexPojo pojo= new IndexPojo();			
			pojo.setStart(Integer.parseInt(split[0]));
			pojo.setEnd(Integer.parseInt(split[1]));
			listIndexOf.add(pojo);
			
		}
		
		tupleFactory = TupleFactory.getInstance();
	}
	

	@SuppressWarnings("rawtypes")
	@Override
	public InputFormat getInputFormat() throws IOException {
		// TODO Auto-generated method stub
		return new TextInputFormat();
	}

	@Override
	public Tuple getNext() throws IOException {
		
		 Tuple tuple = null;
		 @SuppressWarnings("rawtypes")
		List values = new ArrayList();
		  
		
		   boolean notDone;
		try {
			notDone = reader.nextKeyValue();
			
			if (!notDone) {
			       return null;
			   }
			 Text val = (Text) reader.getCurrentValue();
			 
			String line;
				
				if (val == null) {
					return null;
				}
		 
				line = val.toString();
				
				for(IndexPojo indexPojo:listIndexOf){
					
				if(line.length()<indexPojo.getEnd()){
					
					indexPojo.setEnd(line.length());
				}
					
				String subString=line.substring(indexPojo.getStart(), indexPojo.getEnd());	
				
				values.add(subString);
				
					
				}
	
				
			 
			 
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			return null;
		}
		   
		
		tuple = tupleFactory.newTuple(values);
		
		// TODO Auto-generated method stub
		return tuple;
	}

	@Override
	public void prepareToRead(RecordReader reader, PigSplit arg1)
			throws IOException {
		
		this.reader = reader;
		
	}

	@Override
	public void setLocation(String location, Job job) throws IOException {
		
		FileInputFormat.setInputPaths(job, location); 
		
	}
	
	

}
